package com.test.videoplayer.activity.rabbit;

import android.text.TextUtils;
import android.util.Log;

import com.rabbitmq.client.AlreadyClosedException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * Created by zdk on 2019/11/28
 */
@SuppressWarnings("ALL")
public class RabbitMQClient {
    /**
     * 需要自己设置，这里给出的只是样本
     */
    static String SERVICE_HOST_IP;
    static int SERVICE_PORT;
    static String SERVICE_USERNAME;
    static String SERVICE_PASSWORD;

    private static RabbitMQClient singleton;
    private final ConnectionFactory factory;
    private Connection connection;
    private Map<String, Channel> channelMap = new HashMap<>();

    private RabbitMQClient() {
        factory = new ConnectionFactory();

        factory.setHost(SERVICE_HOST_IP);
        factory.setPort(SERVICE_PORT);
        factory.setUsername(SERVICE_USERNAME);
        factory.setPassword(SERVICE_PASSWORD);

        factory.setConnectionTimeout(10000);         //连接时间设置为10秒
        factory.setAutomaticRecoveryEnabled(true);   //恢复连接，通道
        factory.setTopologyRecoveryEnabled(true);    //恢复通道中 转换器，队列，绑定关系等
        factory.setNetworkRecoveryInterval(5000);    //恢复连接间隔，默认5秒

        List<String> exchangeTypeList = new ArrayList<>(4);
        exchangeTypeList.add("fanout"); //不用匹配路由，发送给所有绑定转换器的队列
        exchangeTypeList.add("direct"); //匹配路由一致，才发送给绑定转换器队列
        exchangeTypeList.add("topic");  // 通配符* 和 # 匹配路由一致，才发送给绑定转换器队列
        exchangeTypeList.add("headers");
    }

    static RabbitMQClient getInstance() {
        if (singleton == null) {
            synchronized (RabbitMQClient.class) {
                if (singleton == null) {
                    if (TextUtils.isEmpty(SERVICE_HOST_IP))
                        throw new NullPointerException("请先初始化连接服务端配置条件");
                    singleton = new RabbitMQClient();
                }
            }
        }
        return singleton;
    }


    /**
     * 对应发送：first
     * 不定义转换器（默认匿名装换器），指定队列，获取实时和缓存在队列中的消息
     */
    public void receiveQueueMessage(final String queueName, final ResponseListener listener)
            throws IOException, TimeoutException, AlreadyClosedException {
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    receiveQueueMessageTest(queueName,listener);
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    /**
     * 对应发送：first
     * 自定义非持久化转换器名称和类型，自定义队列，获取实时和缓存在队列中的消息
     */
    public void receiveQueueMessageTest(String queueName, final ResponseListener listener)
            throws IOException, TimeoutException, AlreadyClosedException {
        if (connection == null || !connection.isOpen()) {
            connection = factory.newConnection();
        }
        final Channel channel = connection.createChannel();
        //自定义队列名称，还是匿名队列
        if (TextUtils.isEmpty(queueName)) {
            //队列名称为空的话自己去获取队列名称
            queueName = channel.queueDeclare().getQueue();
        } else {
            channel.queueDeclare(queueName, false, false, false, null);
        }
        //设置绑定
        channel.queueBind(queueName, "test_exchange_topic1", "item.#");
        // 定义队列的消费者
        QueueingConsumer consumer = new QueueingConsumer(channel);
        // 监听队列，手动返回完成
        channel.basicConsume(queueName, false, consumer);
        // 获取消息
        while (true) {
            QueueingConsumer.Delivery delivery = null;
            try {
                delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                Log.d("Tag","所有的 : " + message);
                Thread.sleep(10);
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
       /* //监听队列
        channel.basicConsume(queueName, false, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                String message = new String(body, "UTF-8");
                if (listener != null) {
                    listener.receive(message);
                }
                Log.d("Tag","message : " + message);
                channel.basicAck(envelope.getDeliveryTag(), false);  //消息应答
            }
        });*/
        //String FLAG_RECEIVE = "receive";
        //channelMap.put(FLAG_RECEIVE + queueName, channel);
    }


    /**
     * 关闭所有资源
     */
    public void close() {
        for (Channel next : channelMap.values()) {
            if (next != null && next.isOpen()) {
                try {
                    next.close();
                } catch (IOException | TimeoutException e) {
                    e.printStackTrace();
                }
            }
        }
        channelMap.clear();
        if (connection != null && connection.isOpen()) {
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public interface ResponseListener {
        void receive(String message);
    }
}
